package com.wujian.love.redis;

import com.alibaba.fastjson.JSON;

import com.wujian.love.config.RedisConfig;
import com.wujian.love.contants.ClassContants;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.*;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * redis服务
 */
@Service
public class RedisService {

    private JedisPool jedisPool;

    @Autowired
    private RedisConfig redisConfig;

    /**
     * JedisPool 无法通过@Autowired注入，可能由于是方法bean的原因，此处可以先注入RedisConfig，
     * 然后通过@PostConstruct初始化的时候将factory直接赋给jedisPool
     */
    @PostConstruct
    public void init() {
        jedisPool = redisConfig.redisPoolFactory();
    }


    /**
     * 从redis连接池获取redis实例
     */
    public <T> T get(KeyPrefix prefix, String key, Class<T> clazz) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //对key增加前缀，即可用于分类，也避免key重复
            String realKey = prefix.getPrefix() + key;
            String str = jedis.get(realKey);
            T t = stringToBean(str, clazz);
            return t;
        } finally {
            returnToPool(jedis);
        }

    }

    public <T> T get(String key, Class<T> clazz) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //对key增加前缀，即可用于分类，也避免key重复
            String str = jedis.get(key);
            T t = stringToBean(str, clazz);
            return t;
        } finally {
            returnToPool(jedis);
        }

    }

    /**
     * 存储对象
     */
    public <T> Boolean set(KeyPrefix prefix, String key, T value) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            String str = beanToString(value);
            if (str == null || str.length() <= 0) {
                return false;
            }
            String realKey = prefix.getPrefix() + key;
            int seconds = prefix.expireSeconds();//获取过期时间
            if (seconds <= 0) {
                jedis.set(realKey, str);
            } else {
                jedis.setex(realKey, seconds, str);
            }

            return true;
        } finally {
            returnToPool(jedis);
        }

    }

    /**
     * 存储对象
     */
    public <T> Boolean set(String key, T value) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            String str = beanToString(value);
            if (str == null || str.length() <= 0) {
                return false;
            }
            jedis.set(key, str);
            return true;
        } finally {
            returnToPool(jedis);
        }

    }

    /**
     * 删除
     */
    public boolean delete(KeyPrefix prefix, String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //生成真正的key
            String realKey = prefix.getPrefix() + key;
            long ret = jedis.del(realKey);
            return ret > 0;
        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * 判断key是否存在
     */
    public <T> boolean exists(KeyPrefix prefix, String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //生成真正的key
            String realKey = prefix.getPrefix() + key;
            return jedis.exists(realKey);
        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * 增加值
     * Redis Incr 命令将 key 中储存的数字值增一。    如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作
     */
    public <T> Long incr(KeyPrefix prefix, String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //生成真正的key
            String realKey = prefix.getPrefix() + key;
            return jedis.incr(realKey);
        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * 增加值
     * Redis Incr 命令将 key 中储存的数字值增一。    如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作
     */
    public <T> Long incr(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //生成真正的key
            return jedis.incr(key);
        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * 减少值
     */
    public <T> Long decr(KeyPrefix prefix, String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //生成真正的key
            String realKey = prefix.getPrefix() + key;
            return jedis.decr(realKey);
        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * 减少值
     */
    public <T> Long decr(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.decr(key);
        } finally {
            returnToPool(jedis);
        }
    }

    /***
     * @Author wu
     * @Description //对 key 所储存的字符串值，获取指定偏移量上的位(bit)。当 offset 比字符串值的长度大，或者 key 不存在时，返回 0
     * @Date 15:11 2021/1/13
     * @Param [key, offset]
     * @return void
     **/
    public <T> Boolean getbit(String key, int offset) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.getbit(key, offset);
        } finally {
            returnToPool(jedis);
        }
    }

    public <T> Boolean setbit(String key, int offset, boolean b) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.setbit(key, offset, true);
        } finally {
            returnToPool(jedis);
        }
    }


    public Long bitcount(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.bitcount(key);
        } finally {
            returnToPool(jedis);
        }
    }

    public List<Long> bitfield(String key, String get, String type, String s) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.bitfield(key, "GET", type, "0");
        } finally {
            returnToPool(jedis);
        }
    }

    public static <T> String beanToString(T value) {
        if (value == null) {
            return null;
        }
        Class<?> clazz = value.getClass();
        if (clazz == int.class || clazz == Integer.class) {
            return String.valueOf(value);
        } else if (clazz == long.class || clazz == Long.class) {
            return String.valueOf(value);
        } else if (clazz == String.class) {
            return (String) value;
        } else {
            return JSON.toJSONString(value);
        }

    }

    public static <T> T stringToBean(String str, Class<T> clazz) {
        if (str == null || str.length() <= 0 || clazz == null) {
            return null;
        }
        if (clazz == int.class || clazz == Integer.class) {
            return (T) Integer.valueOf(str);
        } else if (clazz == long.class || clazz == Long.class) {
            return (T) Long.valueOf(str);
        } else if (clazz == String.class) {
            return (T) str;
        } else {
            return JSON.toJavaObject(JSON.parseObject(str), clazz);
        }
    }

    private void returnToPool(Jedis jedis) {
        if (jedis != null) {
            jedis.close();//不是关闭，只是返回连接池
        }
    }


    public StreamEntryID xadd(String redisKey, Map<String, String> map) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            StreamEntryID streamEntryID = jedis.xadd(redisKey, StreamEntryID.NEW_ENTRY, map, 1000, false);
            return streamEntryID;
        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * @return redis.clients.jedis.StreamEntryID
     * @Author wu
     * @Description // 从消息队列中获取消息
     * @Date 17:15 2021/2/2
     * @Param [redisKey]
     **/
    public List<Map.Entry<String, List<StreamEntry>>> xread(String body) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            Map<String, StreamEntryID> map = new HashMap<>();
            StreamEntryID streamEntryID = new StreamEntryID(body);
            map.put("redisKey", streamEntryID);
            List<Map.Entry<String, List<StreamEntry>>> xread = jedis.xread(10, 100, map.entrySet().iterator().next());
            return xread;

        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * @return void
     * @Author wu
     * @Description //创建消费者组
     * @Date 18:17 2021/2/3
     * @Param []
     **/
    public Boolean createConsumerGroup(String key, StreamEntryID entryID) {
        Jedis jedis = null;
        boolean flag = true;
        try {
            jedis = jedisPool.getResource();
            //判断redis 中是否存在组
            List<StreamGroupInfo> streamGroupInfos = jedis.xinfoGroup(key);
            if (streamGroupInfos != null && streamGroupInfos.size() > 0) {
                System.out.println("当前已经存在组");
                flag = false;
            } else {
                String s = jedis.xgroupCreate(key, ClassContants.REDIS_STREAM_GROUP_NAME, null, true);
                System.out.println("创建成功");
                flag = true;
            }
            return flag;

        } finally {
            returnToPool(jedis);
        }
    }

    /**
     * @return void
     * @Author wu
     * @Description //消费消息
     * @Date 18:17 2021/2/3
     * @Param []
     **/
    public Long xack(String key, StreamEntry streamEntry) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            long xack = jedis.xack(key, ClassContants.REDIS_STREAM_GROUP_NAME, streamEntry.getID());
            return xack;
        } finally {
            returnToPool(jedis);
        }
    }

    public void xGetKey(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            Map<String, StreamEntryID> map = new HashMap<>();
            map.put("order-stream", StreamEntryID.LAST_ENTRY);
            Long xlen = jedis.xlen(key);
            System.out.println("长度:" + xlen);

            //获取所有的消息
            List<StreamEntry> xrange = jedis.xrange(key, null, null, 1000);
            for (StreamEntry streamEntry : xrange) {
                StreamEntryID id = streamEntry.getID();
                System.out.println("streamEntry:" + id);
                //long xack = jedis.xack(key, ClassContants.REDIS_STREAM_CONSUMER_GROUP, streamEntry.getID());
                // System.out.println("有没有消费xack:"+xack);
                System.out.println("map:" + streamEntry.getFields().toString());
                // List<Map.Entry<String, List<StreamEntry>>> entries = jedis.xreadGroup(key, "consumner-redisKey", 10, 1000, false, 0-0);
            }
            System.out.println("List size:" + xrange.size());


            //
            List<StreamGroupInfo> streamGroupInfos = jedis.xinfoGroup(key);

            List<StreamConsumersInfo> streamConsumersInfos = jedis.xinfoConsumers(key, ClassContants.REDIS_STREAM_GROUP_NAME);
            if (streamGroupInfos != null) {
                for (StreamGroupInfo streamGroupInfo : streamGroupInfos) {
                    System.out.println("name:" + streamGroupInfo.getName());
                    System.out.println("groupInfo:" + streamGroupInfo.getGroupInfo().toString());
                    System.out.println("lastDeliveredId:" + streamGroupInfo.getLastDeliveredId());
                }
            }


            //
            StreamInfo streamInfo = jedis.xinfoStream(key);
            System.out.println("streamInfo:" + streamInfo.getStreamInfo().toString());

        } finally {
            returnToPool(jedis);
        }
    }


    public List<Map.Entry<String, List<StreamEntry>>> xreadGroup( String groupname,  String consumer,  int count,  long block, boolean noAck,  Map.Entry<String, StreamEntryID> streams) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            List<Map.Entry<String, List<StreamEntry>>> entries = jedis.xreadGroup(groupname, consumer, count, block, noAck, streams);
            return entries;
        } finally {
            returnToPool(jedis);
        }

    }

    public void groupRead() {
        Jedis jedis = jedisPool.getResource();
        String stream = "redisKey";
        Map<String, StreamEntryID> map = new HashMap<>();
        map.put(stream, null);//null 则为 > 重头读起，也可以为$接受新消息，还可以是上一次未读完的消息id
        Map.Entry e = null;
        for(Map.Entry c:map.entrySet()){
            e=c;
        }


//noAck为false的话需要手动ack，true则自动ack. commsumer新建的方式为xreadgroup。
        List<Map.Entry<String, StreamEntryID>> list = jedis.xreadGroup("order-group", "wujian-a", 10, 0, false, e);
        for (Map.Entry m : list) {
            System.out.println(m.getKey() + "---" + m.getValue().getClass());
            if (m.getValue() instanceof ArrayList) {
                List<StreamEntry> l = (List) m.getValue();
                Map<String, String> result = l.get(0).getFields();
                for (Map.Entry entry : result.entrySet()) {
                    System.out.println(entry.getKey() + "---" + entry.getValue());
                }
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                //jedis.xack(stream, group, l.get(0).getID());
                //System.out.println("消息消费成功");
            }
        }


       /* //读取stream中的值
        Map.Entry<String, StreamEntryID> next = map.entrySet().iterator().next();
        List<Map.Entry<String, List<StreamEntry>>> entries = jedis.xreadGroup(ClassContants.REDIS_STREAM_CONSUMER_GROUP, "wujian-a", 10, 1000, false, e);
        for (Map.Entry<String, List<StreamEntry>> stringListEntry : entries) {
            String key = stringListEntry.getKey();
            List<StreamEntry> value = stringListEntry.getValue();
            System.out.println("key:" + key);
            for (StreamEntry streamEntry : value) {
                System.out.println(streamEntry.getFields());
                System.out.println(streamEntry.getID());
            }

        }*/


    }
}
